import java.util.List;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.StructuredTaskScope;
import java.util.stream.Stream;

/**
 * 自定义关闭策略
 *
 * @author Junki
 * @date 2023-11-14 11:06:07
 */
public class Demo04 {

    class MyScope<T> extends StructuredTaskScope<T> {

        private final Queue<T> results = new ConcurrentLinkedQueue<>();

        MyScope() { super(null, Thread.ofVirtual().factory()); }

        @Override
        protected void handleComplete(Subtask<? extends T> subtask) {
            if (subtask.state() == Subtask.State.SUCCESS)
                // 子任务报告完成状态时，将结果加入到results中
                results.add(subtask.get());
        }

        @Override
        public MyScope<T> join() throws InterruptedException {
            super.join();
            return this;
        }

        // 返回成功完成的子任务的结果流
        public Stream<T> results() {
            super.ensureOwnerAndJoined();
            return results.stream();
        }

    }

    <T> List<T> allSuccessful(List<Callable<T>> tasks) throws InterruptedException {
        try (var scope = new MyScope<T>()) {
            for (var task : tasks) scope.fork(task);
            return scope.join()
                    .results().toList();
        }
    }
}
